package com.cx.base.day25_Rocketmq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class BatchConsumer {
    public static void main(String[] args) {
        consumer();
    }

    public static void consumer() {
        //        1 创建消费者Consumer ，并确定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
//        2 指定NameServer地址
        consumer.setNamesrvAddr("192.168.48.128:9876;192.168.48.137:9876");
//        consumer.setNamesrvAddr("106.52.131.197:9876;47.96.165.93:9876");
        try {
//            3 订阅主题Topic和Tag
            consumer.subscribe("batchTopic", "*");
            //            设置订阅模式，有负载均衡、广播两种模式，默认负载均衡,下面设置的是广播
//            consumer.setMessageModel(MessageModel.BROADCASTING);
//        4 设置回调函数，处理消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                //            接受消息内容
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        System.out.println(Thread.currentThread().getName() + new String(msg.getBody()));
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
//        5 启动消息内容
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
